package com.rabiitmq.spring;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Component
public class Listener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.test.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.test.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC
            ),
            key = {"#.#"}))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }

    // -------------------------------- 延时队列关单 -----------------------------------------
    @RabbitListener(queues = "ORDER-CLOSE-QUEUE")// 监听死信队列
    public void closeOrderListener(String orderNo, Channel channel, Message message){
        System.out.println("30分钟后自动关闭订单");
        try {
            // 关单
            // 如果关单失败，说明订单可能已被关闭，直接确认消息
            // 手动ACK
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }catch (Exception e){
            // 消费失败后重新入队
            try {
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
            } catch (IOException ex) {
                log.error("自动关闭订单失败,orderNo=[{}]", orderNo);
            }
        }
    }
    // -------------------------------- 延时队列关单配置信息 -----------------------------------------
    // 发送消息->交换机->（路由）延时队列（消息过期后）->（发送消息到）交换机->（路由）死信队列->监听死信队列
    // 简单说就是：
        // 1.下单发送消息到延时队列，延时队列绑定交换机
        // 2.绑定的交换机发送消息到死信队列(普通队列)，然后监听死信队列(普通队列)
    // 1.延时队列声明
    @Bean("ORDER-TTL-QUEUE")
    public org.springframework.amqp.core.Queue ttlQueue(){
        //延时队列中的消息过期了，会自动触发消息的转发，通过指定routing-key发送到指定exchange中
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange", "spring.test.exchange");
        map.put("x-dead-letter-routing-key", "order.close");
        map.put("x-message-ttl", 1800000); // 单位:毫秒 30分钟仅仅用于测试，实际根据需求，通常30分钟或者15分钟
        return new org.springframework.amqp.core.Queue("ORDER-TTL-QUEUE",true,false,false,map);
    }
    // 2.将延时队列绑定到交换机（就是普通交换机）
    // 订单入库成功后会发送一条routingKey=order.create的消息到spring.test.exchange，发送消息的routingKey与此处的routingKey一致
    @Bean("ORDER-TTL-BINDING")
    public Binding ttlBinding(){
        // 订单入库成功后会发送一条routingKey=order.create的消息到spring.test.exchange,然后会被路由到延时队列ORDER-TTL-QUEUE,延时队列没有消费者,到期后会将消息转发
        return new Binding("ORDER-TTL-QUEUE", Binding.DestinationType.QUEUE, "spring.test.exchange","order.create",null);
    }
    // 3.死信队列 (普通队列)
    @Bean("ORDER-CLOSE-QUEUE")
    public org.springframework.amqp.core.Queue queue(){
        return new org.springframework.amqp.core.Queue("ORDER-CLOSE-QUEUE",true,false,false,null);
    }
    /**
     * 4.死信队列绑定到交换机 并且设置交换机routingKey路由到延时队列
     * routingKey：order.close 和延时队列的routingKey一致,延时队列将消息转发给exchange,exchange再路由到死信队列
     */
    @Bean("ORDER-CLOSE-BINDING")
    public Binding binding(){
        return new Binding("ORDER-CLOSE-QUEUE", Binding.DestinationType.QUEUE,"spring.test.exchange","order.close",null);
    }

    // 定义交换机
//    @Bean("ORDER-CLOSE-EXCHANGE")
//    public org.springframework.amqp.core.Exchange exchange(){
//        return new TopicExchange("order-close-exchange",true,false,null);
//    }
}
